package mr

import (
	"log"
	"sync"
	"time"
)
import "net"
import "net/rpc"
import "net/http"

const (
	MAP    = 1
	REDUCE = 2
	EXIT   = 3
	WAIT   = 4
)

/*自定义队列*/
type queue struct {
	arr   []Task
	mutex sync.Mutex
}

func (this *queue) size() int {
	this.mutex.Lock()
	size := len(this.arr)
	this.mutex.Unlock()
	return size
}
func (this *queue) offer(o Task) {
	this.mutex.Lock()
	this.arr = append(this.arr, o)
	this.mutex.Unlock()
}
func (this *queue) poll() Task {
	this.mutex.Lock()
	task := this.arr[0]
	this.arr = this.arr[1:]
	this.mutex.Unlock()
	return task
}
func (this *queue) remove(id int) {
	this.mutex.Lock()
	for i, v := range this.arr {
		if v.id == id {
			this.arr = append(this.arr[:i], this.arr[i+1:]...)
		}
	}
	this.mutex.Unlock()
}
func (this *queue) contains(task *Task) bool {
	this.mutex.Lock()
	for _, v := range this.arr {
		if v.id == task.id {
			return true
		}
	}
	this.mutex.Unlock()
	return false
}
func (this *queue) getOutTimeTask() []Task {
	this.mutex.Lock()
	outArr := make([]Task, 0)
	for i, task := range this.arr {
		if time.Now().Sub(task.begin) > time.Duration(time.Second*10) {
			outArr = append(outArr, task)
			this.arr = append(this.arr[:i], this.arr[i+1:]...)
		}
	}
	this.mutex.Unlock()
	return outArr
}
func (this *queue) append(tasks []Task) {
	this.arr = append(this.arr, tasks...)
}

type Coordinator struct {
	mapTask          queue //map任务
	mapRunningTask   queue //正在执行的map任务
	reduceTask       queue //需要执行的reduce任务
	reduceRuningTask queue //正在执行的reduce任务
	nReduce          int
	nFiles           int
	fin              bool
	mutex            sync.Mutex
}

type Task struct {
	id    int //任务id
	tType int //Map、Reduce
	begin time.Time
	//state        int      //任务状态,0未启动 1正在执行Map 2 正在执行Reduce 2任务完成 3任务失败
	input        string   //输入文件
	intermediate []string //中间文件位置
	output       string   //reduce任务输出结果
	nReduce      int
	nFiles       int
}

func (this *Task) start() {
	this.begin = time.Now()
}

func TaskToResponse(task *Task, res *Response, state int) {
	res.State = state
	if task != nil {
		res.Id = task.id
		res.Input = task.input
		res.Intermediate = task.intermediate
		res.Output = task.output
		res.NReduce = task.nReduce
		res.NFiles = task.nFiles
	}
}

func (this *Coordinator) Assign(args *Response, reply *Response) error {
	if this.mapTask.size() > 0 { //有空闲map任务
		task := this.mapTask.poll()
		task.start()
		this.mapRunningTask.offer(task)
		TaskToResponse(&task, reply, MAP)
	} else if this.mapRunningTask.size() > 0 { //还有map任务在执行，等待
		reply.State = WAIT
	} else if this.reduceTask.size() > 0 { //map任务都执行完了
		task := this.reduceTask.poll()
		task.start()
		this.reduceRuningTask.offer(task)
		TaskToResponse(&task, reply, REDUCE)
	} else if this.reduceRuningTask.size() > 0 {
		reply.State = WAIT
	} else if this.fin {
		reply.State = EXIT
	}
	return nil
}

/* worker超时检查*/
func (this *Coordinator) timeOutHandle() {
	for {
		if this.fin {
			return
		}
		time.Sleep(time.Second * 10)
		if this.mapRunningTask.size() > 0 {
			mapTasks := this.mapRunningTask.getOutTimeTask()
			if len(mapTasks) > 0 {
				this.mapTask.append(mapTasks)
			}
		}
		if this.reduceRuningTask.size() > 0 {
			reduceTasks := this.reduceRuningTask.getOutTimeTask()
			if len(reduceTasks) > 0 {
				this.reduceTask.append(reduceTasks)
			}
		}
	}
}

func (this *Coordinator) HandleReport(args *Response, reply *Response) error {
	switch args.State {
	case MAP:
		this.mapRunningTask.remove(args.Id)
		if this.mapTask.size() == 0 && this.mapRunningTask.size() == 0 {
			this.assignReduce()
		}
	case REDUCE:
		this.reduceRuningTask.remove(args.Id)
		if this.reduceRuningTask.size() == 0 && this.reduceTask.size() == 0 {
			this.mutex.Lock()
			this.fin = true
			this.mutex.Unlock()
		}
	default:
	}
	return nil
}
func (this *Coordinator) assignReduce() {
	this.reduceTask = queue{}
	this.reduceRuningTask = queue{}
	for i := 0; i < this.nReduce; i++ {
		task := Task{i, REDUCE, time.Time{}, "", nil, "", this.nReduce, this.nFiles}
		this.reduceTask.offer(task)
	}
}

//
// start a thread that listens for RPCs from worker.go
//
func (this *Coordinator) server() {
	rpc.Register(this)
	rpc.HandleHTTP()
	l, e := net.Listen("tcp", ":8080")
	//sockname := coordinatorSock()
	//os.Remove(sockname)
	//l, e := net.Listen("unix", sockname)
	if e != nil {
		log.Fatal("listen error:", e)
	}
	go http.Serve(l, nil)
}

//
// main/mrcoordinator.go calls Done() periodically to find out
// if the entire job has finished.
//
func (this *Coordinator) Done() bool {
	this.mutex.Lock()
	defer this.mutex.Unlock()
	return this.fin
}

//
// create a Coordinator.
// main/mrcoordinator.go calls this function.
// NReduce is the number of reduce tasks to use.
//
func MakeCoordinator(files []string, nReduce int) *Coordinator {
	c := Coordinator{}

	// Your code here.
	c.mapTask = queue{}
	c.mapRunningTask = queue{}
	c.nFiles = len(files)
	c.nReduce = nReduce
	for i, v := range files {
		task := Task{}
		task.tType = MAP
		task.id = i
		task.input = v
		task.nReduce = nReduce
		c.mapTask.offer(task)
	}
	c.server()
	go c.timeOutHandle()
	return &c
}
